package ru.gerunn.silent.message;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;

import ru.gerunn.silent.core.Session;
import ru.gerunn.silent.core.SessionException;
import ru.gerunn.silent.io.WriterException;

public class MessageQueueCallable implements Callable<Object> {

	public static Logger log = Logger.getLogger(MessageQueueCallable.class);

	private final MessageQueue messageQueue;
	private final ConcurrentHashMap<String, Session> sessions;

	public MessageQueueCallable(final MessageQueue messageQueue, final ConcurrentHashMap<String, Session> sessions) {
		this.messageQueue = messageQueue;
		this.sessions = sessions;
	}

	public Object call() throws MessageQueueCallableException {
		MessageQueueElement element = null;
		try {
			while ((element = messageQueue.get()) != null) {
				log.info(element.getSession().getUser().getName() + ": " + element.getMessage());
				for (Session session : sessions.values()) {
					try {
						session.getWriter().write(element.getMessage());
					} catch (WriterException e) {
						try {
							session.close();
						} catch (SessionException e1) {
							throw new MessageQueueCallableException(e1);
						}
						sessions.remove(session.getId());
						throw new MessageQueueCallableException(e);
					}
				}
			}
		} catch (InterruptedException e) {
			throw new MessageQueueCallableException(e);
		}
		return null;
	}

}
